package com.home.fansea.config;

import cn.hutool.json.JSONUtil;
import com.home.fansea.entity.Alarm;
import com.home.fansea.entity.DeviceInfo;
import com.home.fansea.utils.InfluxDBTemplate;
import com.home.fansea.utils.MyMQTTClient;
import com.home.fansea.utils.SpringUtils;
import lombok.AllArgsConstructor;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author WXY
 * @date 2022/6/29 20:43
 */

public class MyMQTTCallback implements MqttCallbackExtended {


    //手动注入
    private MqttConfiguration mqttConfiguration = SpringUtils.getBean(MqttConfiguration.class);

    private static final Logger log = LoggerFactory.getLogger(MyMQTTCallback.class);

//    @Resource
    private MyMQTTClient myMQTTClient;

//    @Resource
    private InfluxDBTemplate influxDBTemplate;
    private static final ExecutorService threadPoolExecutor = Executors.newFixedThreadPool(5);

    public MyMQTTCallback(MyMQTTClient myMQTTClient, InfluxDBTemplate influxDBTemplate) {
        this.myMQTTClient = myMQTTClient;
        this.influxDBTemplate = influxDBTemplate;
    }


   /**
     * 丢失连接，可在这里做重连
     * 只会调用一次
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.error("mqtt connectionLost 连接断开，5S之后尝试重连: {}", throwable.getMessage());
        long reconnectTimes = 1;
        while (true) {
            try {
                if (MyMQTTClient.getClient().isConnected()) {
                    //判断已经重新连接成功  需要重新订阅主题 可以在这个if里面订阅主题  或者 connectComplete（方法里面）  看你们自己选择
                    log.warn("mqtt reconnect success end  重新连接  重新订阅成功");
                    return;
                }
                reconnectTimes+=1;
                log.warn("mqtt reconnect times = {} try again...  mqtt重新连接时间 {}", reconnectTimes, reconnectTimes);
                MyMQTTClient.getClient().reconnect();
            } catch (MqttException e) {
                log.error("mqtt断连异常", e);
            }
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e1) {
            }
        }
    }

    /**
     * @param topic
     * @param mqttMessage
     * @throws Exception
     * subscribe后得到的消息会执行到这里面
     * 接收到的json数据：(deviceInfo)
     * {
     *     deviceId : 1234567890,
     *     temp：1234567890,
     *     hum：1234567890,
     *     luminosity: 1234567890,
     *     time：1234567890
     * }
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        log.info("接收消息主题 : {}，接收消息内容 : {}", topic, new String(mqttMessage.getPayload()));
//        Map maps = (Map) JSON.parse(new String(mqttMessage.getPayload(), CharsetUtil.UTF_8));
        //接收到的json数据：(deviceInfo)
        DeviceInfo deviceInfo = JSONUtil.toBean(new String(mqttMessage.getPayload(), StandardCharsets.UTF_8), DeviceInfo.class);
        //发布消息主题
        if (topic.equals("mqtt/device")){
            //你自己的业务接口
            insertPushAlarm(deviceInfo);
        }
    }

    private void insertPushAlarm(DeviceInfo deviceInfo) {
        //判断是否报警
        if (deviceInfo.getTemp() > 30 || deviceInfo.getHum() > 30 || deviceInfo.getLuminosity() > 30){
            //报警
            log.info("红色预警！！！！");
            Alarm alarm = new Alarm(5, deviceInfo.getTime(), "设备异常！");
            //将alarm转化为json字符串
            String alarmJson = JSONUtil.toJsonStr(alarm);
            myMQTTClient.publish(alarmJson,"mqtt/alarm");
        }
        threadPoolExecutor.execute(new MqttMSgHandle(deviceInfo));
    }

    @AllArgsConstructor
    public class MqttMSgHandle implements Runnable{
        DeviceInfo deviceInfo;
        @Override
        public void run() {
            try {
                //保存到influxDB
                Map<String, String> tag = new HashMap<>();
                tag.put("deviceName",deviceInfo.getDeviceName());
                tag.put("deviceType",deviceInfo.getDeviceType());
                HashMap<String, Object> fields = new HashMap<>();
                fields.put("temp",deviceInfo.getTemp());
                fields.put("hum",deviceInfo.getHum());
                fields.put("luminosity",deviceInfo.getLuminosity());
                influxDBTemplate.write("deviceLog",tag,fields);
                log.info("deviceLog保存到influxDB成功!");
            }catch (Exception e){
                log.error("记录Log异常！",e);
                handlePendingList();
            }
        }
        private void handlePendingList() {
            while (true){
                try {
                    //处理MQTT的数据
                break;
                }catch (Exception e){
                    log.error("出现pending-list订单异常",e);
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException interruptedException) {
                        interruptedException.printStackTrace();
                    }
                }
            }
        }
    }


    /**
     *连接成功后的回调 可以在这个方法执行 订阅主题  生成Bean的 MqttConfiguration方法中订阅主题 出现bug 
     *重新连接后  主题也需要再次订阅  将重新订阅主题放在连接成功后的回调 比较合理
     * @param reconnect
     * @param serverURI
     */
    @Override
    public  void  connectComplete(boolean reconnect,String serverURI){
        log.info("MQTT 连接成功，连接方式：{}",reconnect?"重连":"直连");
        //订阅主题
        myMQTTClient.subscribe(mqttConfiguration.topic1, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic2, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic3, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic4, 1);
    }

    /**
     * 消息到达后
     * subscribe后，执行的回调函数
     *
     * @param s
     * @param mqttMessage
     * @throws Exception
     */
    /**
     * publish后，配送完成后回调的方法
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());
    }
  }

